package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.IndexBean;
import com.alibaba.fastjson.JSON;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.List;

public class SecIndexHbaseFunction implements AllWindowFunction<IndexBean,List<Put>,TimeWindow> {

    @Override
    public void apply(TimeWindow timeWindow, Iterable<IndexBean> iterable, Collector<List<Put>> collector) throws Exception {
        List<Put> list = new ArrayList<>();
        String rowkey=null;
        for (IndexBean indexBean : iterable) {
            rowkey = indexBean.getIndexCode()+indexBean.getTradeTime();

            String str = JSON.toJSONString(indexBean);
            Put put = new Put(Bytes.toBytes(rowkey));
            put.addColumn("info".getBytes(),"data".getBytes(),str.getBytes());
            list.add(put);
        }
        collector.collect(list);
    }
}
